package com.millstein.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSONObject;
import com.millstein.realtime.util.DimUtil;
import com.millstein.realtime.util.DruidDSUtil;
import com.millstein.realtime.util.RedisUtil;
import com.millstein.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;

import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {
    ThreadPoolExecutor threadPool;
    DruidDataSource dataSource;

    /**
     * 待操作的表名
     * @return
     */
    protected abstract String getTableName();

    /**
     * 待操作的id
     * @param input 这个id从哪个实体类中来
     * @return
     */
    protected abstract String getId(T input);

    /**
     * 填充维度信息
     * @param bean 待填充的对象
     * @param dim 维度信息
     */
    protected abstract void supplementDim(T bean, JSONObject dim);

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
        dataSource = DruidDSUtil.getPhoenixConnection();
    }

    @Override
    public void close() throws Exception {
        threadPool.shutdown();
        dataSource.close();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {

        threadPool.submit(new Runnable() {
            Connection connection = null;
            Jedis redisClient = null;

            @Override
            public void run() {
                try {
                    connection = dataSource.getConnection();
                    redisClient = RedisUtil.getRedisClient();
                    JSONObject dim = DimUtil.readDim(
                            connection, redisClient, getTableName(), getId(input)
                    );
                    supplementDim(input, dim);
                    resultFuture.complete(Collections.singletonList(input));
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    if (connection != null) {
                        try {
                            connection.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }
                    if (redisClient != null) {
                        redisClient.close();
                    }
                }

            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("----> " + JSONObject.toJSONString(input));
        super.timeout(input, resultFuture);
    }
}
